package com.hzmg.support.serialization;

import akka.serialization.JSerializer;
import com.esotericsoftware.kryo.kryo5.Kryo;
import com.esotericsoftware.kryo.kryo5.io.Input;
import com.esotericsoftware.kryo.kryo5.io.Output;
import com.esotericsoftware.kryo.kryo5.objenesis.strategy.StdInstantiatorStrategy;
import com.esotericsoftware.kryo.kryo5.util.DefaultInstantiatorStrategy;
import com.esotericsoftware.kryo.kryo5.util.Pool;


import java.io.ByteArrayOutputStream;

public class MessageSerializer extends JSerializer {
    @Override
    public Object fromBinaryJava(byte[] bytes, Class<?> manifest) {
        Kryo kryo = kryoPool.obtain();
        Object o = kryo.readClassAndObject(new Input(bytes));
        kryoPool.free(kryo);

        return o;
    }

    @Override
    public int identifier() {
        return 1037816;
    }

    @Override
    public byte[] toBinary(Object o) {
        Kryo kryo = kryoPool.obtain();

        ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
        Output output = new Output(outputStream);

        if (o == null) {
            kryo.writeClassAndObject(output, o);
        } else {
            kryo.writeObject(output, o);
        }

        kryoPool.free(kryo);

        output.flush();
        byte[] binaryResult = outputStream.toByteArray();
        output.close();

        return binaryResult;
    }

    @Override
    public boolean includeManifest() {
        return false;
    }

    private Pool<Kryo> kryoPool = new Pool<Kryo>(true, false, 6) {
        @Override
        protected Kryo create() {
            Kryo kryo = new Kryo();

            kryo.setRegistrationRequired(false);

            kryo.setInstantiatorStrategy(new DefaultInstantiatorStrategy(
                    new StdInstantiatorStrategy()
            ));

            return kryo;
        }
    };
}
